package io.kiki.sba.registry.server.clientInterface.resource;


import com.google.common.collect.Sets;
import io.kiki.sba.registry.api.exchange.RequestException;
import io.kiki.sba.registry.common.model.Response;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.server.shared.meta.MetaServerService;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

public final class Sdks {
    private static final Logger logger = LoggerFactory.getLogger(Sdks.class);

    private Sdks() {
    }

    public static List<Url> getOtherConsoleServers(String zone, Config config, MetaServerService metaNodeService) {
        if (StringUtils.isBlank(zone)) {
            zone = config.getSessionServerRegion();
        }
        if (StringUtils.isNotBlank(zone)) {
            zone = zone.toUpperCase();
        }
        List<Url> others = new LinkedList<>();
        for (String server : metaNodeService.getSessionServerList(zone)) {
            if (!ServerEnv.isLocalServer(server)) {
                others.add(new Url(server, config.getConsolePort()));
            }
        }
        return others;
    }

    public static Map<Url, Response> concurrentSdkSend(ExecutorService pool, List<Url> servers, SdkExecutor executor, int timeoutMs) {
        Map<Url, Response> responses = Collections.synchronizedMap(new HashMap<>(servers.size() + 1));

        final CountDownLatch latch = new CountDownLatch(servers.size());
        for (Url url : servers) {
            pool.submit(() -> {
                try {
                    Response resp = exec(executor, url);
                    responses.put(url, resp);
                } finally {
                    latch.countDown();
                }
            });
        }
        try {
            boolean success = latch.await(timeoutMs, TimeUnit.MILLISECONDS);
            if (!success) {
                logger.error("[ConcurrentSendError]concurrent send timeout.");
            }

        } catch (InterruptedException e) {
            logger.error("[ConcurrentSendError]concurrent send error.", e);
        } finally {
            Sets.SetView<Url> difference = Sets.difference(new HashSet<>(servers), responses.keySet());
            for (Url url : difference) {
                responses.put(url, Response.buildFailedResponse("execute fail"));
            }
        }
        return responses;
    }

    public static Response getFailedResponseIfAbsent(Collection<Response> responses) {
        if (CollectionUtils.isEmpty(responses)) {
            return Response.buildFailedResponse("response is empty");
        }

        Optional<Response> failedResponses = responses.stream().filter(r -> !r.isSuccess()).findFirst();
        return failedResponses.orElseGet(Response::buildSuccessResponse);
    }

    static Response exec(SdkExecutor executor, Url url) {
        try {
            return executor.execute(url);
        } catch (Throwable e) {
            if (e instanceof RequestException) {
                final Throwable cause = e.getCause();
                // failed on connect
                if (cause.getMessage().contains("connect RemotingException")) {
                    logger.warn("connect failed {}", url, e);
                    return new Response(true, "ignored error: connection failed");
                }
            }
            logger.error("send request other session error!url={}", url, e);
            return new Response(false, e.getMessage());
        }
    }

    public interface SdkExecutor {
        Response execute(Url url) throws Exception;
    }
}
